package com.kingsoft.dc.khaos.optimize

import com.kingsoft.dc.khaos.{KhaosConf, KhaosContext}
import com.kingsoft.dc.khaos.metadata.{ModuleCommon}
import org.json4s.jackson.JsonMethods._
import org.json4s.{DefaultFormats, _}

import scala.collection.mutable


/**
 * Created by chengguangqing on 2019/7/31.
 */
object DagOptimize {

  def appendDeleteModuleWithCos(kc: KhaosContext): ModuleCommon = {

    val moduleConf = "{\"operator_name\":\"map\",\"clazz_type\":\"transform\",\"id\":\"999999\",\"name\":\"ClearMppOnCosData\",\"engine\":\"SPARK\",\"clazz\":\"com.kingsoft.dc.khaos.module.spark.preprocess.transform.Transform\",\"type\":\"transform\",\"strategy\":{\"clazz\":" +
      "\"com.kingsoft.dc.khaos.module.spark.preprocess.specific.ClearMppOnCosData\",\"config\":{}}}"

    kc.conf.set("999999", moduleConf)
    val module = new ModuleCommon( //封装
      "999999",
      "built-in-delete",
      "SPARK",
      "com.kingsoft.dc.khaos.module.spark.preprocess.transform.Transform",
      "ACTION",
      moduleConf, //moduleConf
      "",
      Nil,
      Nil,
      Nil,
      Nil
    )
    module
  }

  def appendDeleteModuleWithRedis(kc: KhaosContext): ModuleCommon = {

    val moduleConf = "{\"operator_name\":\"map\",\"clazz_type\":\"transform\",\"id\":\"777777\",\"name\":\"ClearRedisSyncData\",\"engine\":\"SPARK\",\"clazz\":\"com.kingsoft.dc.khaos.module.spark.preprocess.transform.Transform\",\"type\":\"transform\",\"strategy\":{\"clazz\":" +
      "\"com.kingsoft.dc.khaos.module.spark.preprocess.specific.ClearRedisSyncData\",\"config\":{}}}"

    kc.conf.set("777777", moduleConf)
    val module = new ModuleCommon( //封装
      "777777",
      "built-in-delete",
      "SPARK",
      "com.kingsoft.dc.khaos.module.spark.preprocess.transform.Transform",
      "ACTION",
      moduleConf, //moduleConf
      "",
      Nil,
      Nil,
      Nil,
      Nil
    )
    module
  }

  def appendOffsetModuleWithKafka(kc: KhaosContext): ModuleCommon = {
    val moduleConf = "{\"operator_name\":\"map\",\"clazz_type\":\"transform\",\"id\":\"888888\",\"name\":\"ManageOffset\",\"engine\":\"SPARK\",\"clazz\":\"com.kingsoft.dc.khaos.module.spark.preprocess.transform.Transform\",\"type\":\"transform\",\"strategy\":{\"clazz\":" +
      "\"com.kingsoft.dc.khaos.module.spark.preprocess.specific.ManageOffset\",\"config\":{}}}"

    kc.conf.set("888888", moduleConf)
    val module = new ModuleCommon( //封装
      "888888",
      "kafka-manage-offset",
      "SPARK",
      "com.kingsoft.dc.khaos.module.spark.preprocess.transform.Transform",
      "ACTION",
      moduleConf, //moduleConf
      "",
      Nil,
      Nil,
      Nil,
      Nil
    )
    module
  }

  def appendCollectModuleWithAppEnd(kc: KhaosContext): ModuleCommon = {
    val moduleConf = "{\"operator_name\":\"map\",\"clazz_type\":\"transform\",\"id\":\"666666\",\"name\":\"CollectMetrics\",\"engine\":\"SPARK\",\"clazz\":\"com.kingsoft.dc.khaos.module.spark.preprocess.transform.CollectMetrics\",\"type\":\"transform\",\"strategy\":{\"clazz\":" +
      "\"com.kingsoft.dc.khaos.module.spark.preprocess.specific.CollectMetrics\",\"config\":{}}}"

    kc.conf.set("666666", moduleConf)
    val module = new ModuleCommon( //封装
      "666666",
      "collect-metrics",
      "SPARK",
      "com.kingsoft.dc.khaos.module.spark.preprocess.transform.Transform",
      "ACTION",
      moduleConf, //moduleConf
      "",
      Nil,
      Nil,
      Nil,
      Nil
    )
    module
  }

  //替换算子，初步实现，后期需要把json放到配置文件中，还有参数
  def replaceVirtualModule(json: String, conf: KhaosConf): JValue = {

    var actual = "{\"id\":\"${id}\",\"name\":\"检核上报-ES\",\"operator_name\":\"checkSink\",\"clazz_type\":\"sink\",\"type\":\"sink\",\"engine\":\"SPARK\",\"clazz\":\"com.kingsoft.dc.khaos.module.spark.sink.Sink\",\"strategy\":{\"clazz\":\"com.kingsoft.dc.khaos.module.spark.sink.ESCheckSink\",\"config\":{\"db_name\":\"ksc.dm.qu.integration\",\"table_name\":\"quality\",\"doc_id\":{\"gen_rule_name\":\"uuid\",\"rule\":\"\"},\"extract_fields\":[{\"field\":\"taskFlowNum\",\"data_type\":\"STRING\",\"from_field\":\"\",\"field_props\":{\"not_null\":\"true\",\"key_type\":\"\",\"default_value\":\"\"}},{\"field\":\"ruleId\",\"data_type\":\"STRING\",\"from_field\":\"\",\"field_props\":{\"not_null\":\"true\",\"key_type\":\"\",\"default_value\":\"\"}},{\"field\":\"taskSubmitTime\",\"data_type\":\"STRING\",\"from_field\":\"\",\"field_props\":{\"not_null\":\"true\",\"key_type\":\"\",\"default_value\":\"\"}},{\"field\":\"tenantId\",\"data_type\":\"STRING\",\"from_field\":\"\",\"field_props\":{\"not_null\":\"true\",\"key_type\":\"\",\"default_value\":\"\"}},{\"field\":\"dataDetaile\",\"data_type\":\"STRING\",\"from_field\":\"\",\"field_props\":{\"not_null\":\"true\",\"key_type\":\"\",\"default_value\":\"\"}},{\"field\":\"errorDescribe\",\"data_type\":\"STRING\",\"from_field\":\"\",\"field_props\":{\"not_null\":\"false\",\"key_type\":\"\",\"default_value\":\"\"}}]}}}"
    implicit val formats = DefaultFormats
    val parser = parse(json, true)

    //获取参数
    //val ip= conf.getString("")

    val widgets: JValue = (parser \ "graph" \ "widgets")
    val newWidgets = mutable.ListBuffer[JObject]()

    widgets.children.foreach { module =>
      var newModule = module
      val moduleMap = module.values.asInstanceOf[Map[String, _]]
      val clazzType = String.valueOf(moduleMap.get("type").get)

      if (clazzType.equals("virtual")) {
        val id = String.valueOf(moduleMap.get("id").get)
        val operName = String.valueOf(moduleMap.get("operator_name").get)
        //检核算子
        if (operName.equals("checkSink")) {
          //根据信息替换模板
          actual = actual.replace("${id}", id)
          val act = parse(actual, true)
          newModule = act
        }
        //其他的替换
        if (operName.equals("other")) {

        }
      }
      newWidgets.append(newModule.asInstanceOf[JObject])
    }
    JArray(newWidgets.toList)
  }


  def main(args: Array[String]): Unit = {
    val aaa = "{\"graph\":{\"edges\":[{\"destination\":{\"node\":\"47\",\"seq\":\"0\"},\"source\":{\"node\":\"46\",\"seq\":\"0\"}},{\"destination\":{\"node\":\"48\",\"seq\":\"0\"},\"source\":{\"node\":\"47\",\"seq\":\"0\"}}],\"widgets\":[{\"operator_name\":\"mpp\",\"clazz_type\":\"source\",\"engine\":\"SPARK\",\"name\":\"source_1564711827316\",\"id\":\"46\",\"type\":\"source\",\"strategy\":{\"clazz\":\"com.kingsoft.dc.khaos.module.spark.source.MppSource\",\"config\":{\"filter\":\"\",\"extender\":{\"auth\":{\"params\":{\"ds_name\":\"mpp\",\"user_id\":\"1\",\"project_id\":8,\"ds_type\":\"mpp\",\"ds_id\":105},\"clazz\":\"com.kingsoft.dc.khaos.extender.meta.impl.DmAuthImpl\"},\"meta\":{\"params\":{\"ds_name\":\"mpp\",\"user_id\":\"1\",\"project_id\":8,\"ds_type\":\"mpp\",\"ds_id\":105},\"clazz\":\"com.kingsoft.dc.khaos.extender.meta.impl.MppDmMetaImpl\"}},\"db_name\":\"test_mzq\",\"table_name\":\"mysql_mzq_t1\",\"advanced_options\":{},\"extract_fields\":[{\"field\":\"a1\",\"data_type\":\"STRING\",\"length\":\"50\",\"colCode\":\"col-21202F2938212B3E22272626252E434D\",\"techLogicTableId\":672,\"id\":2891,\"position\":0},{\"field\":\"a2\",\"data_type\":\"NUMBER\",\"length\":\"9\",\"colCode\":\"col-68955f3a03e4439c84a4eede0e017c24\",\"techLogicTableId\":672,\"id\":2898,\"position\":0},{\"field\":\"a3\",\"data_type\":\"DECIMAL\",\"length\":\"15,2\",\"colCode\":\"col-48f73dad2f3d489eb44cf1b8f2ddab5f\",\"techLogicTableId\":672,\"id\":2905,\"position\":0},{\"field\":\"a4\",\"data_type\":\"DATE\",\"colCode\":\"col-714311fcbdd242d3ab6f0e51f360b952\",\"techLogicTableId\":672,\"id\":2912,\"position\":0},{\"field\":\"a5\",\"data_type\":\"DATETIME\",\"colCode\":\"col-349361b8f9ef4cf3b3e9480f2f2f9c0f\",\"techLogicTableId\":672,\"id\":2919,\"position\":0}]}},\"clazz\":\"com.kingsoft.dc.khaos.module.spark.source.Source\"},{\"operator_name\":\"map\",\"clazz_type\":\"transform\",\"engine\":\"SPARK\",\"name\":\"map_1564711841579\",\"id\":\"47\",\"type\":\"transform\",\"strategy\":{\"clazz\":\"com.kingsoft.dc.khaos.module.spark.preprocess.transform.NewFilterAndMap\",\"config\":{\"filter\":\"\",\"advanced_options\":{\"partition\":{\"on_off\":false,\"nums\":\"\"},\"cache_option\":{\"on_off\":false,\"option\":\"cache\"}},\"extract_fields\":[{\"field\":\"a1\",\"data_type\":\"STRING\",\"length\":\"50\",\"tag\":\"source_column\",\"option\":\"\"},{\"field\":\"a2\",\"data_type\":\"NUMBER\",\"length\":\"9\",\"tag\":\"source_column\",\"option\":\"\"},{\"field\":\"a3\",\"data_type\":\"DECIMAL\",\"length\":\"15,2\",\"tag\":\"source_column\",\"option\":\"\"},{\"field\":\"a4\",\"data_type\":\"DATE\",\"tag\":\"source_column\",\"option\":\"\"},{\"field\":\"a5\",\"data_type\":\"DATETIME\",\"tag\":\"source_column\",\"option\":\"\"}]}},\"clazz\":\"com.kingsoft.dc.khaos.module.spark.preprocess.transform.Transform\"},{\"id\":\"48\",\"name\":\"检核上报\",\"type\":\"virtual\",\"operator_name\":\"checkSink\"}]}}"
    val cc = DagOptimize.replaceVirtualModule(aaa, null)
    val moduleConf = compact(render(cc))
    println(moduleConf)
  }
}
